package com.vhall.component.service.room.impl;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.vhall.component.dao.room.RecordOfRoomMapper;
import com.vhall.component.dao.room.RoomExtendsMapper;
import com.vhall.component.dao.room.RoomJoinsMapper;
import com.vhall.component.dao.room.RoomMapper;
import com.vhall.component.entity.record.RecordEntity;
import com.vhall.component.entity.room.dto.RoomsEntityDTO;
import com.vhall.component.entity.room.entity.RoomExtendsEntity;
import com.vhall.component.entity.room.entity.RoomJoinsEntity;
import com.vhall.component.entity.room.entity.RoomsEntity;
import com.vhall.component.entity.room.enums.RoomAuditStatusEnum;
import com.vhall.component.entity.room.enums.RoomAuditWhichEnum;
import com.vhall.component.entity.room.enums.RoomStatusEnum;
import com.vhall.component.framework.common.constant.CommonConstant;
import com.vhall.component.framework.common.exception.BusinessException;
import com.vhall.component.plugin.common.annotation.RedisSpinLock;
import com.vhall.component.plugin.common.constant.HttpConstant;
import com.vhall.component.plugin.common.constant.PaasCommonConstant;
import com.vhall.component.plugin.common.constant.RedisKey;
import com.vhall.component.plugin.common.exception.BizErrorCode;
import com.vhall.component.plugin.core.service.PaasService;
import com.vhall.component.service.room.LiveService;
import com.vhall.component.service.room.RoomCacheService;
import com.vhall.component.service.room.RoomInnerService;
import com.vhall.component.service.room.concant.RoomConstant;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * @author rui.zhou
 * @date 2021/4/28 16:50
 */
@Slf4j
@Service
public class LiveServiceImpl implements LiveService {

    private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern(CommonConstant.DATETIME_FORMAT);

    @Autowired
    private PaasService paasService;

    @Autowired
    private RoomMapper roomMapper;

    /**
     * 现在的逻辑是paas只调用这一次的回调，然后在这个回调中去生成回放，这样会导致record服务和room服务发生循环依赖。
     * 所以在这里直接操作record表
     */
    @Autowired
    private RecordOfRoomMapper recordOfRoomMapper;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    private RoomInnerService roomInnerService;

    @Autowired
    private RoomExtendsMapper roomExtendsMapper;

    @Value("${vhall.lottery.stream-name:lottery-stream}")
    public String streamName;

    @Autowired
    private RoomCacheService roomCacheService;

    @Autowired
    private RoomJoinsMapper roomJoinsMapper;
    private static DateTimeFormatter df = DateTimeFormatter.ofPattern(CommonConstant.DATETIME_FORMAT);
    /**
     * 是否开启房间审核
     * 1为开启
     * 0为关闭房间审核
     */
    @Value("${project.room.audit:0}")
    public Integer roomAuditWitch;
    @Override
    public String callBack(Map<String, Object> callBack) {
        String resp = "success";
        log.info("开播或者结束直播回调" + callBack);
        String event = (String) callBack.get("event");
        if (com.vhall.component.plugin.common.constant.RoomConstant.TELL_STREAM.equals(event)) {
            return resp;
        }
        String roomId = (String) callBack.get("room_id");
        RoomsEntityDTO roomsEntityDTO = roomInnerService.getRoomInfo(roomId);

        if (roomsEntityDTO == null) {
            log.error("没有查询到房间");
            resp = "fail";
            return resp;
        }
        //拉取paas流状态列表
        JSONObject result = paasService.getStreamStatus(roomId);
        if (result.getInteger(HttpConstant.CODE) != HttpConstant.SUCCESS_CODE) {
            log.error("拉取paas流状态列表错误");
            resp = "fail";
            return resp;
        }
        JSONObject data = result.getJSONObject(HttpConstant.DATA);
        JSONObject obj = data.getJSONObject(roomId);
        String pushTime = obj.getString("push_time");
        String endTime = obj.getString("end_time");
        if (obj.getInteger(PaasCommonConstant.STREAM_STATUS) == com.vhall.component.plugin.common.constant.RoomConstant.STATUS_START) {
            startRoomLive(roomId, roomsEntityDTO, pushTime);
        }
        if (obj.getInteger(PaasCommonConstant.STREAM_STATUS) == com.vhall.component.plugin.common.constant.RoomConstant.STATUS_STOP) {
            // 直播结束，更新该房间下的，聊天总数
            JSONObject msgCount = paasService.getMessageCount(roomId);
            if (msgCount.getInteger(HttpConstant.CODE) != HttpConstant.SUCCESS_CODE) {
                log.error("获取聊天总数错误");
                resp = "fail";
                return resp;
            }
            stopRoomLive(roomId, roomsEntityDTO, pushTime, endTime, msgCount);
        }
        return resp;
    }

    private void stopRoomLive(String roomId, RoomsEntityDTO roomsEntityDTO, String pushTime, String endTime, JSONObject msgCount) {

        roomInnerService.updateByIl(RoomsEntityDTO.builder()
                .ilId(roomsEntityDTO.getIlId())
                .messageTotal(msgCount.getInteger("count"))
                .endLiveTime(LocalDateTime.parse(endTime, df)).build());

        /**
         * 如果当前直播的状态已经是停止播放,则说明是手动停止,获取手动开始时的直播时间,用于生成完整回放
         */
        if (com.vhall.component.plugin.common.constant.RoomConstant.STATUS_STOP == roomsEntityDTO.getStatus()) {
            LocalDateTime beginLiveTime = roomsEntityDTO.getBeginLiveTime();
            if (Objects.nonNull(beginLiveTime)) {
                pushTime = beginLiveTime.format(df);
            }
            redisTemplate.delete(RedisKey.ROOM_LIVE_START_CHANGE_STATUS_KEY + roomId);
        }
        //直播结束之后 马上 将当前 直播间的连接数清理为0
        redisTemplate.opsForValue().set(RedisKey.CONNECT_COUNT_BY_ROOM + roomsEntityDTO.getIlId(), 0);
        //直播结束之后  将账号下面对应的 房间id  清除 set
        redisTemplate.opsForSet().remove(RedisKey.LIVING_ROOMS_OF_ACCOUNT + roomsEntityDTO.getAccountId(), roomsEntityDTO.getIlId());

        redisTemplate.opsForHash().delete(RedisKey.INTERACT_GLOBAL + roomId, com.vhall.component.plugin.common.constant.RoomConstant.START_TYPE);

        // 这个地方目前没有使用，这个字段用来存储直播模式。所以这里不再使用。
        // 创建回放放在更新房间信息后面，为了减少直播结束时间的延迟
        createRecord(roomsEntityDTO, pushTime, endTime);
    }

    /**
     * 房间直播开始回调处理
     *
     * @param roomId         房间 id
     * @param roomsEntityDTO 房间信息
     * @param pushTime       开始推流时间
     */
    private void startRoomLive(String roomId, RoomsEntityDTO roomsEntityDTO, String pushTime) {
        boolean obsFlag = false;
        RoomsEntity startRoom = new RoomsEntity();
        if (RoomAuditWhichEnum.OPEN_ROOM_AUDIT.getStatus().equals(roomAuditWitch)
                && !RoomAuditStatusEnum.APPROVED.getAuditStatus().equals(roomsEntityDTO.getAuditStatus())) {
            throw new BusinessException(BizErrorCode.COMP_ROOM_AUDIT_STATUS_ERROR);
        }
        //不是开播中  代表是 obs推流
        if (roomsEntityDTO.getStatus() != com.vhall.component.plugin.common.constant.RoomConstant.STATUS_START) {
            obsFlag = true;
            createRoomExtend(roomId);
            redisTemplate.opsForHash().put(RedisKey.INTERACT_GLOBAL + roomId, com.vhall.component.plugin.common.constant.RoomConstant.START_TYPE, "7");
            startRoom.setBeginLiveTime(LocalDateTime.parse(pushTime, df));
        }
        startRoom.setIlId(roomsEntityDTO.getIlId());
        startRoom.setStatus(com.vhall.component.plugin.common.constant.RoomConstant.STATUS_START);
        setRoomBeginLiveTime(roomId, pushTime, startRoom);
        log.info("开播更新状态开始");
        roomMapper.updateById(startRoom);
        //填充房间扩展信息
        redisTemplate.delete(RedisKey.ROOM_INFO_BY_ROOM_ID + roomsEntityDTO.getRoomId());
        redisTemplate.delete(RedisKey.ROOM_INFO_BY_ROOM_ID + roomsEntityDTO.getIlId());

        // 开播之后 把对应的账号id 加到 app并发控制级别里面
        redisTemplate.opsForSet().add(RedisKey.LIVING_ROOMS_OF_APPS, roomsEntityDTO.getAccountId());

        //开始直播之后把 当前直播间 id 加入到 账号下的直播房间set   主要 用于做并发控制
        redisTemplate.opsForSet().add(RedisKey.LIVING_ROOMS_OF_ACCOUNT + roomsEntityDTO.getAccountId(), roomsEntityDTO.getIlId());

        // OBS推流  发送消息通知观看端房间已开播
        if (obsFlag) {
            JSONObject json = new JSONObject();
            json.put("type", com.vhall.component.plugin.common.constant.RoomConstant.LIVE_START);
            json.put(RoomConstant.ROOM_JOIN_ID, roomsEntityDTO.getAccountId());
            //房间消息
            paasService.sendMessage(roomsEntityDTO.getChannelId(), json.toJSONString());
            //通知消息
            paasService.sendMessage(roomsEntityDTO.getNifyChannel(), json.toJSONString());
        }
        log.info("开播更新状态完毕");
    }

    /**
     * 设置房间的开始直播时间
     *
     * @param roomId    房间id
     * @param pushTime  推流时间
     * @param startRoom 房间对象
     */
    private void setRoomBeginLiveTime(String roomId, String pushTime, RoomsEntity startRoom) {
        /**
         * 直播间状态的中间态,用于辨别直播间开播时,直播是否首次回调
         * 首次回调时状态为-待直播:RoomConstant.STATUS_WAITING
         */
        Object roomMidStatus = redisTemplate.opsForValue().get(RedisKey.ROOM_LIVE_START_CHANGE_STATUS_KEY + roomId);
        if (Objects.nonNull(roomMidStatus)) {
            int status = Integer.parseInt(roomMidStatus.toString());
            if (status == RoomStatusEnum.TO_BE_BROADCAST.getStatus()) {
                startRoom.setBeginLiveTime(LocalDateTime.parse(pushTime, df));
                redisTemplate.opsForValue().set(RedisKey.ROOM_LIVE_START_CHANGE_STATUS_KEY + roomId, com.vhall.component.plugin.common.constant.RoomConstant.STATUS_START, 2, TimeUnit.DAYS);
            }
        } else {
            startRoom.setBeginLiveTime(LocalDateTime.parse(pushTime, df));
            redisTemplate.opsForValue().set(RedisKey.ROOM_LIVE_START_CHANGE_STATUS_KEY + roomId, com.vhall.component.plugin.common.constant.RoomConstant.STATUS_START, 2, TimeUnit.DAYS);
        }
    }


    public RoomJoinsEntity getHostRoomJoin(String roomId) {
        RoomJoinsEntity entity = roomCacheService.getCachedHostRoomJoin(roomId);
        if (Objects.isNull(entity)) {
            LambdaQueryWrapper<RoomJoinsEntity> wrapper = Wrappers.lambdaQuery();
            wrapper.eq(RoomJoinsEntity::getRoomId, roomId);
            wrapper.eq(RoomJoinsEntity::getRoleName, com.vhall.component.plugin.common.constant.RoomConstant.HOST)
                    .last(RoomConstant.LIMIT_1);
            entity = roomJoinsMapper.selectOne(wrapper);
            // 将主持人参会信息单独缓存
            roomCacheService.cacheHostRoomJoin(entity);
        }
        return entity;
    }


    @RedisSpinLock(redisKey = "#cacheKey", expireTime = 10 * 1000)
    public RoomExtendsEntity cacheAndHandleGetRoomExtend(String roomId) {
        //从缓存获取 房间扩展信息
        RoomExtendsEntity roomExtendsEntity = roomCacheService.getRoomExtendCache(roomId);
        if (Objects.isNull(roomExtendsEntity)) {
            LambdaQueryWrapper<RoomExtendsEntity> wrapper = Wrappers.lambdaQuery();
            wrapper.eq(RoomExtendsEntity::getRoomId, roomId);
            wrapper.eq(RoomExtendsEntity::getEndType, 0);
            wrapper.orderByDesc(RoomExtendsEntity::getCreatedAt);
            wrapper.last(" limit 1 ");
            roomExtendsEntity = roomExtendsMapper.selectOne(wrapper);
            if (Objects.isNull(roomExtendsEntity)) {
                return null;
            }
            //放入缓存
            roomCacheService.cacheRoomExtend(roomExtendsEntity);
        }
        return roomExtendsEntity;
    }

    private void createRoomExtend(String roomId) {
        RoomExtendsEntity entity = new RoomExtendsEntity();
        entity.setRoomId(roomId);
        LocalDateTime now = LocalDateTime.now();
        entity.setStartTime(now);
        entity.setCreatedAt(now);
        entity.setUpdatedAt(now);
        entity.setStartType(7);
        roomExtendsMapper.insert(entity);
    }

    /**
     * 创建 回放
     *
     * @param roomsEntityDTO
     * @param endTime
     */
    private void createRecord(RoomsEntityDTO roomsEntityDTO, String pushTime, String endTime) {
        //2、获取生成信息
        JSONObject vod = paasService.createRecord(roomsEntityDTO.getRoomId(), pushTime, endTime);
        if (vod.getInteger(HttpConstant.CODE) != HttpConstant.SUCCESS_CODE) {
            log.error("添加直播回放生成任务错误,param:[{},{},{}],result:[{}]",
                    roomsEntityDTO.getRoomId(), pushTime, endTime, vod.toJSONString());
        } else {
            String vodId = vod.getJSONObject(HttpConstant.DATA).getString(PaasCommonConstant.VOD_ID_KEY);
            RecordEntity recordEntity = new RecordEntity();
            recordEntity.setRoomId(roomsEntityDTO.getRoomId());
            recordEntity.setAccountId(roomsEntityDTO.getAccountId());
            recordEntity.setIlId(roomsEntityDTO.getIlId());
            recordEntity.setSource(0);
            recordEntity.setVodId(vodId);
            recordEntity.setName(roomsEntityDTO.getRoomId() + " " + pushTime);
            //获取视频详情
            JSONObject recordInfo = paasService.getRecordInfo(vodId);
            if (recordInfo.getInteger(HttpConstant.CODE) != HttpConstant.SUCCESS_CODE) {
                log.error("获取视频详细信息错误:param:[{}], result:[{}]", vodId, recordInfo.toJSONString());
            } else {
                JSONObject vodInfo = recordInfo.getJSONObject(HttpConstant.DATA).getJSONObject("vod_info");
                recordEntity.setCreatedTime(LocalDateTime.parse(vodInfo.getString("created_at"), FORMATTER));
                recordEntity.setTranscodeStatus(vodInfo.getInteger("transcode_status"));
                recordEntity.setStorage(vodInfo.getInteger("storage"));
                recordEntity.setDuration(vodInfo.getInteger("duration"));
                recordEntity.setRoomId(vodInfo.getString("room_id"));

                recordEntity.setUpdatedAt(LocalDateTime.now());
                recordOfRoomMapper.insert(recordEntity);
                log.info("保存回放成功");
            }
        }
    }
}
